//
// Created by benxb on 2021/11/27.
//

#include "storage/data_chunk_io.h"
#include <cstdio>
#include <compress/compressor.h>
#include <zlib.h>

namespace benxdb {


DataChunkIO::DataChunkIO(const std::string& table_name) :
data_storage_(table_name), desc_storage_(table_name), bitmap_storage_(table_name) {
    // init free_chunk_list
    for (auto & i : chunk_buffer_pool_) {
        free_chunk_list_.push_back(&i);
    }
}

bool DataChunkIO::AppendDataChunk(DataChunk* data_chunk, bool need_compress) {
    ChunkDesc desc;
    unsigned long out_size = data_chunk->GetChunkSize();
    auto *compressed_data = data_chunk->GetData();
    if (GlobalConfig::compress_strategy == DELTA) {
        Compressor compressor;
        std::string out;
        compressor.DeltaEncode(data_chunk->GetData(), data_chunk->GetChunkSize(), data_chunk->GetDataType(), out);
        compressed_data = (char *) malloc(out.size());
        memcpy(compressed_data, out.data(), out.size());
        out_size = out.size();
    } else if (GlobalConfig::compress_strategy == ZLIB) {
        compressed_data = (char *) malloc(data_chunk->GetChunkSize());
        out_size = data_chunk->GetChunkSize();
        compress(reinterpret_cast<Bytef *>(compressed_data), &out_size,
                 reinterpret_cast<const Bytef *>(data_chunk->GetData()), out_size);
    } else if (GlobalConfig::compress_strategy == DELTA_ZLIB) {
        Compressor compressor;
        std::string mid_data;
        compressor.DeltaEncodeFloatWithPadding(data_chunk->GetData(), CHUNK_SIZE, 8, mid_data);
        compressed_data = (char *) malloc(mid_data.size());
        out_size = mid_data.size();
        compress(reinterpret_cast<Bytef *>(compressed_data), &out_size,
                 reinterpret_cast<const Bytef *>(mid_data.data()), out_size);
    }

    desc.chunk_size = out_size;
    desc.chunk_offset = data_storage_.GetNextDataChunkOffset();
    desc.compress_mode = GlobalConfig::compress_strategy;
    desc.type = data_chunk->GetDataType();
    data_chunk->GetMaxMinValue(desc.max_val, desc.min_val);
    data_storage_.AppendChunkData(reinterpret_cast<const char *>(compressed_data), out_size);
    desc_storage_.AppendDesc(desc);
    free(compressed_data);
    LOG_DEBUG("compress before: %d, after: %lu", CHUNK_SIZE, out_size);
    return true;
}

bool DataChunkIO::DeleteRecords(const std::vector<uint64_t>& record_ids) {
    bitmap_storage_.DeleteBatch(record_ids);
    return true;
}

DataChunk* DataChunkIO::ReadDataChunk(chunk_id_t chunk_id) {
    if (loaded_chunk_map_.find(chunk_id) != loaded_chunk_map_.end()) {
        return loaded_chunk_map_[chunk_id];
    }
    ChunkDesc desc;
    if (!desc_storage_.GetDescByChunkId(chunk_id, desc)) {
        return nullptr;
    }
    DataChunk* data_chunk = nullptr;
    if (free_chunk_list_.empty()) {
        LOG_ERROR("No free chunk.");
        return data_chunk;
    }
    data_chunk = free_chunk_list_.front();
    free_chunk_list_.pop_front();
    loaded_chunk_map_.insert({chunk_id, data_chunk});
    data_storage_.ReadChunkData(desc.chunk_offset, desc.chunk_size, data_chunk->GetData());
    data_chunk->SetDataType(desc.type);
    data_chunk->SetChunkSize(desc.chunk_size);
    data_chunk->SetChunkId(chunk_id);

    // uncompress
    if (desc.compress_mode == DELTA) {
        Compressor compressor;
        std::string out;
        compressor.DeltaDecode(data_chunk->GetData(), desc.chunk_size, data_chunk->GetDataType(), out);
        memcpy(data_chunk->GetData(), out.data(), out.size());
        data_chunk->SetChunkSize(out.size());
    }
    return data_chunk;
}

void DataChunkIO::UnPin(chunk_id_t chunk_id) {
    if (loaded_chunk_map_.find(chunk_id) == loaded_chunk_map_.end()) {
        return;
    }
    DataChunk* data_chunk = loaded_chunk_map_[chunk_id];
    loaded_chunk_map_.erase(chunk_id);
    free_chunk_list_.push_back(data_chunk);
}

bool DataChunkIO::ReadDataChunkDesk(chunk_id_t chunk_id, ChunkDesc &desc) {
    if (!desc_storage_.GetDescByChunkId(chunk_id, desc)) {
        return false;
    }
    return true;
}

void DataChunkIO::DeleteTableFile() {
    remove(data_storage_.GetDataFile().data());
    remove(bitmap_storage_.BitmapFile().data());
    remove(desc_storage_.GetDescFile().data());
}

void DataChunkIO::ExpandDeleteMap() {
    bitmap_storage_.AppendChunk();
}


}